package com.bclz.cat.uniquekey;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Iterator;
import java.util.SortedMap;
import java.util.TreeMap;

/**
 * 对所有数据归并，求TopN
 */
public class CatReducer extends Reducer<NullWritable,Text,NullWritable,Text> {




    @Override
    protected void reduce(NullWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        SortedMap<Integer,Text> tree=new TreeMap<>((a, b)->b-a);
        Iterator<Text> iterator = values.iterator();
        int N = context.getConfiguration().getInt("hadoop.top.n", 10);

        while(iterator.hasNext()){
            //values对象实现是ValueIterable ，next()方法每次都返回同一个对象,ReduceContextImpl源码 中的value属性
            //使用过程中如果需要保存每次的值，需要clone该对象
            String next = iterator.next().toString();
            String[] split = next.split(",");

            if(split.length>2){
                tree.put(Integer.valueOf(split[2]),new Text(next));
                if(tree.size()>N){
                    tree.remove(tree.lastKey());
                }
            }

        }
        for (Text t:
                tree.values()) {
            context.write(NullWritable.get(),t);

        }

    }


}
